package com.jiming.webrpc;

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;

import com.jiming.webrpc.sys.ProcessObject;
import com.jiming.webrpc.utiles.ThreadUtile;
import com.jiming.webrpc.utiles.TokenUtile;

public class LocalHost extends Host {
	private static LocalHost localHost = null;
	
	private SessionRequestHandler sessionRequestHandler = null;
	
	private static String webrpc_server_ip = "server.webrpc.cn";//39.107.38.234
	
	private static int webrpc_server_port = 19889;
	

	
	
	private int max_pool_size = 100;//缓存池中的消息超过这个数就会被销毁早期数据
	
	private String token;
	
	private DatagramSocket socket = null;
	
	
	private Thread receive_worker = null;
	
	private Thread heartbeat_worker = null;
	
	private boolean isConnection = false;//是否已经和服务建立链接？
	
	/**
	 * 系统消息缓存池
	 */
	private Map<String, Msg> sys_msg_pool = new LinkedHashMap<>();
	
	/**
	 * 组消息池<组id, LinkedHashMap<消息id, Msg>>
	 */
	private Map<String, LinkedHashMap<String, Msg>> group_msg_pool = new HashMap<String, LinkedHashMap<String, Msg>>();
	
	/**
	 * 会话消息池<会话id, LinkedHashMap<消息id, Msg>>
	 */
	private Map<String, LinkedHashMap<String, Msg>> session_msg_pool = new HashMap<String, LinkedHashMap<String, Msg>>();
	
	/**
	 * 缓存用过的InetAddress，避免每次发包都创建，很消耗资源
	 */
	private Map<String, InetAddress> inetAddress_pool = new HashMap<>();
	
	/**
	 * 缓存远程主机
	 */
	private Map<String, RemoteHost> remoteHost_pool = new HashMap<>();
	
	
	private LocalHost() {}
	
	
	public int getLocalPort() {
		return localPort;
	}


	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}


	public String getToken() {
		return token;
	}


	public void setToken(String token) {
		this.token = token;
	}


	public static String getWebrpc_server_ip() {
		return webrpc_server_ip;
	}


	public static void setWebrpc_server_ip(String webrpc_server_ip) {
		LocalHost.webrpc_server_ip = webrpc_server_ip;
	}


	public static int getWebrpc_server_port() {
		return webrpc_server_port;
	}


	public static void setWebrpc_server_port(int webrpc_server_port) {
		LocalHost.webrpc_server_port = webrpc_server_port;
	}


	public SessionRequestHandler getSessionRequestHandler() {
		return sessionRequestHandler;
	}


	public void setSessionRequestHandler(SessionRequestHandler sessionRequestHandler) {
		this.sessionRequestHandler = sessionRequestHandler;
	}


	public static LocalHost newInstance(String token, String email, SessionRequestHandler sessionRequestHandler) throws Exception {
		if(localHost == null) {
			localHost = new LocalHost();
			localHost.setSessionRequestHandler(sessionRequestHandler);
			
			//验证token长度必须大于16位，必须包含大写小写和数字，不能有其他特俗字符串 否则就抛出WebRpcException
			checkToken(token);
			localHost.setToken(token);
			
			//start udp socket
			localHost.socket = new DatagramSocket(10086);
			localHost.setLocalPort(localHost.socket.getPort());
			
		}
		
		return localHost;
	}


	/**
	 * 开始接受线程(异步接收udp包)
	 */
	private void startReceive() {
		if(this.receive_worker != null) {
			this.receive_worker.interrupt();
			this.receive_worker = null;
		}
		
		this.receive_worker = new Thread(() ->{
			while(true) {
				try {
					byte[] data = new byte[DataPackage.dp_max_size];//每个包不能大于11k
		            DatagramPacket dp = new DatagramPacket(data, data.length);
					this.socket.receive(dp);
					
					//分发消息
					doMsg(new String(dp.getData(), 0, dp.getLength(), "UTF-8"));
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			
		});
		
		this.receive_worker.start();
	}
	
	/**
	 * 本机是否已经和服务器建立链接？注册成功返回true
	 * @return
	 */
	public boolean isConnection() {
		return this.isConnection;
	}
	
	private void doMsg(String msgJson) {
		Msg msg = Msg.newBuiledByJson(msgJson);
		if(msg != null) {
			if(MsgLev.sys == msg.getLev()) {
				System.out.println("收到消息:" + msgJson);
				if(Method.start_session == msg.getBody().getMethod()) {
					new Thread(() -> {
						//如果是通知创建会话的消息
						if(localHost.isConnection()) {
							String data = msg.getBody().getData();
							System.out.println("远程主机请求建立session:" + data);
							String[] params = data.split(",");
							if(params.length == 2) {
								String param = params[0];
								String customerAToken = param.split("=")[1];
								
								String param2 = params[1];
								String customerASessionId = param2.split("=")[1];
								
								RemoteHost remoteAHost = localHost.getRemoteHost(customerAToken);
								System.out.println("收到请求建立连接,ip=" + remoteAHost.getNatIp() + ",port=" + remoteAHost.getNatPort());
								SessionStatus sessionStatus = null;
								if(remoteAHost.isOnline()) {
									//可以和A建立会话
									sessionStatus = remoteAHost.accptSessionA(customerASessionId);
								}else {
									System.out.println("对方不在线，放弃连接");
									sessionStatus = SessionStatus.closed;
								}
								
								//把结果响应给服务器
								String redata = "sessionStatus=" + sessionStatus.name() + ",sessionId=" + remoteAHost.getSessionId();
								Msg sessionMsg = Msg.newBuiled().setLev(MsgLev.sys).setId(msg.getId()).setBody(MsgBody.newBuiled().setMethod(Method.response).setData(redata));
								this.sendPackage(sessionMsg, this.webrpc_server_ip, this.webrpc_server_port);
								System.out.println("把start session结果响应给服务器redata:" + redata);
							}
							
						}
					}).start();
					
				}else {
					//删除一直没有被消费的消息
					delOlderMsg(this.sys_msg_pool);
					this.sys_msg_pool.put(msg.getId(), msg);
				}
				
			}else if(MsgLev.group == msg.getLev()) {
				delOlderMsg(getGroupPoolById(msg.getGroupId()));
				addGroupMsg(msg);
			}else if(MsgLev.session == msg.getLev()) {
				delOlderMsg(getSessionById(msg.getSessionId()));
				addSessionMsg(msg);
			}
		}
	}


	private void addSessionMsg(Msg msg) {
		LinkedHashMap<String, Msg> session = this.session_msg_pool.get(msg.getSessionId());
		if(session == null) {
			session = new LinkedHashMap<String, Msg>();
			this.session_msg_pool.put(msg.getSessionId(), session);
		}
		System.out.println("收到session消息:" + msg);
		session.put(msg.getId(), msg);
	}


	private void addGroupMsg(Msg msg) {
		LinkedHashMap<String, Msg> group = this.group_msg_pool.get(msg.getGroupId());
		if(group == null) {
			group = new LinkedHashMap<String, Msg>();
			this.group_msg_pool.put(msg.getGroupId(), group);
		}
		
		group.put(msg.getId(), msg);
	}


	private Map<String, Msg> getSessionById(String sessionId) {
		return this.session_msg_pool.get(sessionId);
	}


	private Map<String, Msg> getGroupPoolById(String groupId) {
		return this.group_msg_pool.get(groupId);
	}


	private void delOlderMsg(Map<String, Msg> msg_pool) {
		if(msg_pool == null) {
			return;
		}
		
		int size = msg_pool.size() - this.max_pool_size;
		if(size > 0) {
			//依次删除最早进来的消息
			int i = 1;
			Iterator<String> keyIt = sys_msg_pool.keySet().iterator();
			while(keyIt.hasNext()) {
				String id = keyIt.next();
				msg_pool.remove(id);
				
				i++;
				if(i > size) {
					break;//结束删除
				}
			}
		}
	}


	public void start() {
		//开始接受线程(异步接收udp包)
		startReceive();
		
		//将本地的token注册到server上
		register();
		
		if(this.isConnection()) {
			//开启心跳
			startHeartbeat();
		}
	}
	
	/**
	 * 心跳
	 */
	private void startHeartbeat() {
		heartbeat_worker = new Thread(() ->{
			while(true) {
				try {
					String id = UUID.randomUUID().toString();
					String data = "token=" + this.token + ",localPort=" + this.getLocalPort();//token=123123123,localPort=10086
					MsgBody body = MsgBody.newBuiled().setMethod(Method.heartbeat).setData(data).builed();
					Msg sysMsg = Msg.newBuiled().setId(id).setLev(MsgLev.sys).setBody(body).builed();
					
					//发送出去udp等待响应
					RpcResult result = sendPackageAndReturn(sysMsg, this.webrpc_server_ip, this.webrpc_server_port);
					
					if((RpcResultStatus.succes == result.getStatus())) {
						String registerResult = result.getMsg().getBody().getData();
						ProcessObject processObject = ProcessObject.newBuiled(registerResult);
						if(com.jiming.webrpc.sys.ProcessResult.reseive_success == processObject.getProcessResult()) {
							//发送心跳成功
						}
						
					}
				}catch (Exception e) {
				}
				
				//秒钟发送一次
				ThreadUtile.sleep(3 * 1000);
			}
			
		});
		heartbeat_worker.start();
	}



	/**
	 * 重启
	 */
	public void reStart() {
		close();
		start();
	}
	
	/**
	 * 关闭
	 */
	public void close() {
		this.isConnection = false;
		
		if(this.socket != null) {
			this.socket.close();
			this.socket = null;
		}
		
		if(this.receive_worker != null) {
			this.receive_worker.interrupt();
			this.receive_worker = null;
		}
		if(this.heartbeat_worker != null) {
			this.heartbeat_worker.interrupt();
			this.heartbeat_worker = null;
		}
		
		
		//关闭所有远程主机
		remoteHost_pool.forEach((token, remoteHost)->{
			remoteHost.close();
		});
		remoteHost_pool.clear();
		
	}


	/**
	 * 验证token长度必须大于16位，必须包含大写小写和数字，不能有其他特俗字符串 否则就抛出WebRpcException
	 * @param token
	 */
	private static void checkToken(String token) {
		TokenUtile.checkToken(token);
	}

	/**
	 * 将token登记到webrpc-server上
	 */
	private void register() {
		String id = UUID.randomUUID().toString();
		String data = "token=" + this.token + ",localPort=" + this.getLocalPort();//token=123123123,localPort=10086
		MsgBody body = MsgBody.newBuiled().setMethod(Method.register).setData(data).builed();
		Msg sysMsg = Msg.newBuiled().setId(id).setLev(MsgLev.sys).setBody(body).builed();
		
		//发送出去udp等待响应
		RpcResult result = sendPackageAndReturn(sysMsg, this.webrpc_server_ip, this.webrpc_server_port);
		
		if((RpcResultStatus.succes == result.getStatus())) {
			String registerResult = result.getMsg().getBody().getData();
			ProcessObject processObject = ProcessObject.newBuiled(registerResult);
			if(com.jiming.webrpc.sys.ProcessResult.register_succes == processObject.getProcessResult()) {
				//注册成功
				Customer local = this.getCustomerByToken(this.token);
				this.setNatIp(local.getNatIp());
				this.setNatPort(local.getNatPort());
				this.isConnection = true;
				
				try {
					Thread.sleep(4000);
				} catch (InterruptedException e) {
				}
			}else {
				//注册失败
				throw new WebRpcException(result.getMsg().getError());
			}
			
		}else {
			//注册失败
			throw new WebRpcException(result.getMsg().getError());
		}
	}

	/**
	 * 发送msg不需要等待回复，只发送一次，没有重试机制，可能会丢包
	 * @param msg
	 * @param host
	 * @param port
	 */
	public void sendPackage(Msg sysMsg, String tarIp, int tarPort) {
		try {
			String msg = sysMsg.toString();
			byte[] data = msg.getBytes("UTF-8");
			
			//数据不能超过11k
			if(data.length > (DataPackage.dp_max_size)) {
				return;
			}
			
			//发送
			DatagramPacket udpPackage = new DatagramPacket(data, data.length, getInetAddress(tarIp), tarPort);
			this.socket.send(udpPackage);
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}

	/**
	 * 发送消息包并等待回应(超时没有回应的尝试3次，否则就算发送失败)
	 * @param sysMsg
	 * @return
	 */
	public RpcResult sendPackageAndReturn(Msg sysMsg, String tarIp, int tarPort) {
		try {
			String msg = sysMsg.toString();
			byte[] data = msg.getBytes("UTF-8");
			
			//数据不能超过11k
			if(data.length > (DataPackage.dp_max_size)) {
				return RpcResult.newBuiled().setStatus(RpcResultStatus.failed).setError("One data package must lt "+DataPackage.dp_max_size+".").builed();
			}
			
			//发送
			DatagramPacket udpPackage = new DatagramPacket(data, data.length, getInetAddress(tarIp), tarPort);
			this.socket.send(udpPackage);
			
			//等待响应（1秒钟超时，最多重试3次，否则超时失败）
			Msg resultMsg = null;
			boolean flag = false;
			for(int i = 0; i < 3; i++) {
				long start = System.currentTimeMillis();
				while((System.currentTimeMillis() - start) < 1000) {
					if(MsgLev.sys == sysMsg.getLev()) {
						resultMsg = this.sys_msg_pool.remove(sysMsg.getId());
						if(resultMsg != null) {
							//收到
							flag = true;
							break;
						}
					}else if(MsgLev.group == sysMsg.getLev()) {
						resultMsg = this.takeGroupMsgById(sysMsg.getGroupId(), sysMsg.getId());
						if(resultMsg != null) {
							//收到
							flag = true;
							break;
						}
					}else if(MsgLev.session == sysMsg.getLev()) {
						resultMsg = this.takeSessionMsgById(sysMsg.getSessionId(), sysMsg.getId());
						if(resultMsg != null) {
							//收到
							flag = true;
							break;
						}
					}
					
					Thread.sleep(1);
				}
				
				if(flag) {
					//已经收到回复
					break;
				}else {
					//超时没收到，重试
					this.socket.send(udpPackage);
				}
			}
			
			
			//最后处理结果
			if(resultMsg == null) {
				return RpcResult.newBuiled().setStatus(RpcResultStatus.timeout).setError("Send udp package timeout.").builed();
			}
			
			//收到响应，说明目标已经收到次udp package,把响应的msg返回出去
			return RpcResult.newBuiled().setStatus(RpcResultStatus.succes).setMsg(resultMsg).builed();
		} catch (Exception e) {
			return RpcResult.newBuiled().setStatus(RpcResultStatus.error).setError(e.getMessage()).builed();
		}
	}


	private Msg takeSessionMsgById(String sessionId, String id) {
		LinkedHashMap<String, Msg> session = session_msg_pool.remove(sessionId);
		if(session == null) {
			session = new LinkedHashMap<String, Msg>();
			session_msg_pool.put(sessionId, session);
		}
		
		return session.remove(id);
	}


	public Msg takeGroupMsgById(String groupId, String id) {
		LinkedHashMap<String, Msg> group = group_msg_pool.remove(groupId);
		if(group == null) {
			group = new LinkedHashMap<String, Msg>();
			group_msg_pool.put(groupId, group);
		}
		return group.remove(id);
	}


	/**
	 * 缓存用过的InetAddress
	 * @param tarIp
	 * @return
	 * @throws UnknownHostException
	 */
	private InetAddress getInetAddress(String tarIp) throws UnknownHostException {
		InetAddress address = this.inetAddress_pool.get(tarIp);
		if(address == null) {
			address = InetAddress.getByName(tarIp);
			this.inetAddress_pool.put(tarIp, address);
		}
		return address;
	}


	/**
	 * 从服务器上获取远程主机的信息
	 * @param remoteToken
	 * @return
	 */
	public RemoteHost getRemoteHost(String remoteToken) {
		RemoteHost remoteHost = remoteHost_pool.get(remoteToken);
		if(remoteHost != null) {
			return remoteHost;
		}
		remoteHost = RemoteHost.newBuild(this);
		Customer remoteCustomer = getCustomerByToken(remoteToken);
		
		if(remoteCustomer != null) {
			//设置远程主机基本信息(成功查询到远程主机)
			remoteHost.setNatIp(remoteCustomer.getNatIp());
			remoteHost.setNatPort(remoteCustomer.getNatPort());
			remoteHost.setLocalPort(remoteCustomer.getLocalPort());
			remoteHost.setActiveTime(remoteCustomer.getActiveTime());
			remoteHost.setToken(remoteToken);
		}
		
		remoteHost_pool.put(remoteToken, remoteHost);
		return remoteHost;
	}
	
	/**
	 * @param token
	 * @return
	 */
	public Customer getCustomerByToken(String token) {
		String id = UUID.randomUUID().toString();
		MsgBody body = MsgBody.newBuiled().setMethod(Method.get_remote).setData("token=" + token).builed();
		Msg sysMsg = Msg.newBuiled().setId(id).setLev(MsgLev.sys).setBody(body).builed();
		
		//发送出去udp等待响应
		RpcResult result = sendPackageAndReturn(sysMsg, this.webrpc_server_ip, this.webrpc_server_port);
		
		if((RpcResultStatus.succes == result.getStatus())) {
			String registerResult = result.getMsg().getBody().getData();
			try {
				Customer remoteCustomer = Customer.newBuiled(registerResult);
				
				//设置远程主机基本信息(成功查询到远程主机)
				
				return remoteCustomer;
			}catch (Exception e) {
			}
		}
		return null;
	}


	/**
	 * 取出会话中所有的消息
	 */
	public List<Msg> takeSessionAllRequestMsg(String sessionId) {
		List<Msg> requests = new ArrayList<Msg>();
		Map<String, Msg> session = this.getSessionById(sessionId);
		if(session == null) {
			return requests;
		}
		
		List<String> ids = new ArrayList<String>();
		Iterator<String> it  = session.keySet().iterator();
		while(it.hasNext()) {
			ids.add(it.next());
		}
		
		//取出所有request消息
		for(String id:ids) {
			Msg msg = session.get(id);
			if(msg != null && (Method.request == msg.getBody().getMethod())) {
				requests.add(msg);
				session.remove(id);
			}
		}
		
		return requests;
	}

	
}
